package com.wy.sloth.core;

import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.ZSetOperations;

import com.alibaba.fastjson.JSONObject;
import com.wy.sloth.config.SlothConfig;
import com.wy.sloth.constant.SlothConstants;
import com.wy.sloth.exception.SlothException;
import com.wy.sloth.model.SlothMessage;
import com.wy.sloth.model.SlothReceipt;
import com.wy.sloth.model.SlothReceiptData;
import com.wy.sloth.redis.service.SlothRedisService;
import com.wy.sloth.util.CollectionUtils;
import com.wy.sloth.util.StringUtils;
import com.wy.sloth.util.ThreadUtils;

import lombok.Getter;
import lombok.Setter;

/**
 * 消息发送类
 * 
 * @author wangyuan
 * @since 2019年1月31日
 */
public class SlothStandardSender implements SlothSender {
    
    private static final Logger logger = LoggerFactory.getLogger(SlothStandardSender.class);
    
    @Getter
    @Setter
    private ListOperations<String, JSONObject> listOperations;
    
    @Getter
    @Setter
    private HashOperations<String, String, JSONObject> hashOperations;
    
    @Getter
    @Setter
    private ZSetOperations<String, String> zSetOperations;
    
    /**
     * 项目关于redis的封装操作服务类
     */
    private SlothRedisService slothRedisService;
    
    /**
     * 项目相关配置项
     */
    @Getter
    @Setter
    private SlothConfig slothConfig;
    
    /**
     * 当前项目名称生成器
     */
    private SlothNameManager slothNameManager;
    
    /**
     * 随机数生成器，用户随机获取消息该发送到哪个消息队列
     */
    private static Random random = new Random();
    
    public SlothStandardSender(RedisTemplate<String, JSONObject> redisTemplate, 
            RedisTemplate<String, String> stringRedisTemplate, 
            SlothConfig slothConfig){
        this.listOperations = redisTemplate.opsForList();
        this.zSetOperations = stringRedisTemplate.opsForZSet();
        this.hashOperations = redisTemplate.opsForHash();
        this.slothConfig = slothConfig;
        
        this.slothNameManager = new SlothStandardNameManager(this.slothConfig);
        this.slothRedisService = new SlothRedisService(redisTemplate, stringRedisTemplate, this.slothConfig);
    }

    @Override
    public SlothReceipt send(String type, JSONObject data) {
        SlothMessage message = new SlothMessage(type, data);
        return send(message);
    }

    @Override
    public SlothReceipt send(SlothMessage message) {
        if(message == null){
            throw new SlothException("消息主体不能为空");
        }
        
        if(StringUtils.isEmpty(message.getType())){
            throw new SlothException("type不能为空");
        }
        
        //根据type获取可以发送的消息队列
        //判断消息是群发消息还是单用户消息
        List<String> msgQueues = slothRedisService.listQueueNameByType(message.getType());
        if(CollectionUtils.isEmpty(msgQueues)){
            throw new SlothException("暂无可用队列接收处理此消息，请确定存在可用的消息处理者");
        }
        
        SlothReceipt msgReceipt = createSlothReceipt(message.isSync());
        if(message.isSingle()){
            //单应用消息，此消息只能由一个应用进行处理
            String msgQueueName = msgQueues.get(random.nextInt(msgQueues.size()));
            beforePushOfEveryQueue(message, msgQueueName, msgReceipt);
            Long responseResult = listOperations.leftPush(msgQueueName, message.toJsonObject());
            afterPushOfEveryQueue(responseResult, message, msgQueueName, msgReceipt, 0);
        } else {
            //多应用消息，此消息由多个可处理此类型消息的应用进行处理
            Long responseResult = null;
            for(int i = 0;i < msgQueues.size();i++){
            	beforePushOfEveryQueue(message, msgQueues.get(i), msgReceipt);
                responseResult = listOperations.leftPush(msgQueues.get(i), message.toJsonObject());
                afterPushOfEveryQueue(responseResult, message, msgQueues.get(i), msgReceipt, i);
            }
        }
        
        //等待回执
        waitReceipt(message, msgReceipt);
        
        return msgReceipt;
    }
    
    /**
     * 推送消息至队列之前执行的动作
     * @param message
     * @param msgQueueName
     * @param msgReceipt
     */
    private void beforePushOfEveryQueue(SlothMessage message, String msgQueueName, SlothReceipt msgReceipt) {
    	slothRedisService.logMsg(message, msgQueueName, SlothConstants.STATUS_NEW);
    	
    	if(message.isSync()){
            msgReceipt.addReceipt(message.getId(), msgQueueName);
        }
    }
    
    /**
     * 推送消息至队列之后执行的动作
     * @param responseResult
     * @param message
     * @param msgQueueName
     * @param msgReceipt
     * @param i
     */
    private void afterPushOfEveryQueue(Long responseResult, SlothMessage message, String msgQueueName, SlothReceipt msgReceipt, int i) {
    	if(responseResult == null) {
        	logger.error("推送消息(id={})至队列{}失败", message.getId(), msgQueueName);
        	slothRedisService.updateDealStatusOfMsgForLog(message.getId(), msgQueueName, SlothConstants.STATUS_EXEC_FAIL, "插入执行队列失败");
        	if(message.isSync()){
                msgReceipt.get(i).setStatus(SlothReceipt.STATUS_ERROR);
                msgReceipt.get(i).setErrorMsg("插入执行队列失败");
            }
        } else {
        	logger.info("推送消息(id={})至队列{}的响应结果为：{}", message.getId(), msgQueueName, responseResult);
        }
    }
    
    private SlothReceipt createSlothReceipt(boolean sync){
        if(sync){
            return new SlothReceipt();
        } else {
            return null;
        }
    }
    
    private void waitReceipt(SlothMessage message, SlothReceipt msgReceipt){
        if(!message.isSync()){
            return;
        }
        
        logger.info("-----开始等待消息（id={}）的消息回执-----", message.getId());
        long startWaitTime = System.currentTimeMillis();
        do {
            JSONObject response = null;
            SlothReceiptData responseInst = null;
            for(SlothReceiptData item : msgReceipt){
                if(item.getStatus() == null){
                    try {
                        response = hashOperations.get(slothNameManager.getMsgReceiptKey(), item.getHashKey());
                        if(response == null){
                            continue;
                        }
                        
                        responseInst = JSONObject.toJavaObject(response, SlothReceiptData.class);
                        item.setStatus(responseInst.getStatus());
                        item.setReceiptData(responseInst.getReceiptData());
                        item.setErrorMsg(responseInst.getErrorMsg());
                        
                        hashOperations.delete(slothNameManager.getMsgReceiptKey(), item.getHashKey());
                    } catch (Exception e) {
                        logger.error("获取消息回执时出现错误，消息ID=" + item.getMsgId() + "，原因是：", e);
                    }
                }
            }
            
            if(System.currentTimeMillis() - startWaitTime > message.getMaxWaitTime()){
                for(SlothReceiptData item : msgReceipt){
                    if(item.getStatus() == null){
                        item.setStatus(SlothReceipt.STATUS_ERROR);
                        item.setReceiptData(null);
                        item.setErrorMsg("等待消息回执超时");
                    }
                }
                break;
            }
        } while(msgReceipt.existUnanswered() && ThreadUtils.sleep(TimeUnit.MILLISECONDS, 200L));
        logger.info("-----结束等待消息（id={}）的消息回执-----", message.getId());
    }
}
